import numpy as np # linear algebra
import pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)

# import gensim.utils as g_utils
import re
from nltk.tokenize.treebank import TreebankWordDetokenizer
import gensim.utils as gutils

# Loading the Emotion Dataset
df = pd.read_csv('/kaggle/input/emotion-dataset/emotion-dataset.csv')

df.head()

# Displaying Descriptive Statistics
df.describe()

def preprocess(data):
    
    # Removing URLs with a regular expression
    url_pattern = re.compile(r'https?://\S+|www\.\S+')
    data = url_pattern.sub(r'', data)

    # Remove emails
    data = re.sub('\S*@\S*\s?', '', data)

    # Remove new line characters
    data = re.sub('\s+', ' ', data)

    # Remove distracting single quotes
    data = re.sub("\'", "", data)
        
    return data

def sent_2_words(sentences):
    """
    Tokenizes and preprocesses a list of sentences using Gensim's simple_preprocess.
    
    Parameters:
    - sentences: List of sentences to be tokenized.
    
    Returns:
    - A generator object yielding lists of words for each sentence.
    """
    for sentence in sentences:
        yield gutils.simple_preprocess(str(sentence), deacc=True)

def de_tokenize(text):
    """
    Detokenizes a list of words using TreebankWordDetokenizer.
    
    Parameters:
    - text: List of words to be detokenized.
    
    Returns:
    - Detokenized sentence.
    """
    return TreebankWordDetokenizer().detokenize(text)

from sklearn.preprocessing import LabelEncoder

# Performing Label Encoding for Emotions
df['emotion_label'] = LabelEncoder().fit_transform(df['emotion'])

df.head()

df['emotion'].value_counts().plot(kind='bar')

df[df['text'].duplicated() == True]

# Removing Duplicate Rows from the DataFrame
index = df[df['text'].duplicated() == True].index
df.drop(index, axis = 0, inplace = True)
df.reset_index(inplace=True, drop = True)

df[df['text'].duplicated() == True]

len(df)

# Splitting the Dataset into Train and Test Sets
X_train = np.array(df['text'].values.tolist()[:15000])
X_test = np.array(df['text'].values.tolist()[15000:])
y_train = np.array(df['emotion'].values.tolist()[:15000])
y_test = np.array(df['emotion'].values.tolist()[15000:])

len(X_train) == len(y_train)

from sklearn.model_selection import train_test_split,KFold, GridSearchCV
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.tree import DecisionTreeClassifier
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score,confusion_matrix, classification_report
from sklearn.pipeline import Pipeline
from sklearn.metrics import f1_score
from sklearn.svm import SVC
from sklearn.feature_extraction.text import TfidfVectorizer

def train_model(model, data, targets):
    """
    Trains a text classification model using scikit-learn's Pipeline.

    Parameters:
    - model: The machine learning classifier to be trained.
    - data: Input features (text data).
    - targets: Corresponding labels.

    Returns:
    - A trained Pipeline object containing TfidfVectorizer and the given model.
    """
    # Create a Pipeline object with a TfidfVectorizer and the given model
    text_clf = Pipeline([('vect', TfidfVectorizer()),
                         ('clf', model)])
    # Fit the model on the data and targets
    text_clf.fit(data, targets)
    return text_clf

#Train the model with the training data
log_reg = train_model(LogisticRegression(solver='liblinear',random_state = 0), X_train, y_train)

#test the model with the test data
y_pred=log_reg.predict(X_test)

#calculate the accuracy
log_reg_accuracy = accuracy_score(y_test, y_pred)
print('Accuracy: ', log_reg_accuracy,'\n')

# Calculate and Display F1 Scores for Each Emotion
f1score = f1_score(y_test,y_pred, average=None)
pd.DataFrame(f1score, index=df.emotion.unique(), columns=['F1 Scores'])

# Generate a detailed Classification Report
print(classification_report(y_test, y_pred))

# Train the model with the training data
dec_tree = train_model(DecisionTreeClassifier(random_state=0), X_train, y_train)

# Test the model with the test data
y_pred = dec_tree.predict(X_test)

# Calculate the accuracy
DTC_accuracy = accuracy_score(y_test, y_pred)
print('Accuracy: ', DTC_accuracy, '\n')

# Calculate and Display F1 Scores for Each Emotion
f1score = f1_score(y_test, y_pred, average=None)
pd.DataFrame(f1score, index=df.emotion.unique(), columns=['F1 Scores'])

# Generate a detailed Classification Report
print(classification_report(y_test, y_pred))

#Train the model with the training data
SVM = train_model(SVC(random_state = 0), X_train, y_train)

#test the model with the test data
y_pred=SVM.predict(X_test)

#calculate the accuracy
SVM_accuracy = accuracy_score(y_test, y_pred)
print('Accuracy: ', SVM_accuracy,'\n')

f1score = f1_score(y_test,y_pred, average=None)
pd.DataFrame(f1score, index=df.emotion.unique(), columns=['F1 Scores'])

print(classification_report(y_test, y_pred))

models = pd.DataFrame({
    'Model': ['Logistic Regression', 'Decision Tree','Support Vector Machine'],
    'Accuracy': [log_reg_accuracy.round(2), DTC_accuracy.round(2), SVM_accuracy.round(2)]})

models.sort_values(by='Accuracy', ascending=False).reset_index().drop(['index'], axis=1)

import seaborn as sns
import nltk
#Lime
from lime import lime_text
from lime.lime_text import LimeTextExplainer
from lime.lime_text import IndexedString,IndexedCharacters
from lime.lime_base import LimeBase
from lime.lime_text import explanation
sns.set(font_scale=1.3)
nltk.download('omw-1.4')

df.emotion.unique()

dec_tree.classes_

X_test[15]

np.array(df['emotion'].values.tolist()[15015])

#c_LR = make_pipeline(tfidf, clf)
explainer_LR = LimeTextExplainer(class_names=dec_tree.classes_)
idx  = 56
print("Actual Text : ", X_test[idx])
print("Prediction : ", dec_tree.predict(X_test)[idx])
print("Actual :     ", y_test[idx])
exp = explainer_LR.explain_instance(X_test[idx], dec_tree.predict_proba,top_labels=5)
exp.show_in_notebook()

custom_x = np.array(['i am not sad, but happy, yes not very sad'])
#c_LR = make_pipeline(tfidf, clf)
explainer_LR = LimeTextExplainer(class_names=dec_tree.classes_)
idx  = 56
print("Actual Text : ", custom_x)
print("Prediction : ", dec_tree.predict(custom_x)[0])
print("Actual :     ", 'joy')
exp = explainer_LR.explain_instance(custom_x[0], dec_tree.predict_proba,top_labels=5)
exp.show_in_notebook()

#c_LR = make_pipeline(tfidf, clf)
explainer_LR = LimeTextExplainer(class_names=dec_tree.classes_)
idx  = 15
print("Actual Text : ", X_test[idx])
print("Prediction : ", dec_tree.predict(X_test)[idx])
print("Actual :     ", y_test[idx])
exp = explainer_LR.explain_instance(X_test[idx], dec_tree.predict_proba,top_labels=5)
exp.show_in_notebook()

#c_LR = make_pipeline(tfidf, clf)
explainer_LR = LimeTextExplainer(class_names=dec_tree.classes_)
idx  = 157
print("Actual Text : ", X_test[idx])
print("Prediction : ", dec_tree.predict(X_test)[idx])
print("Actual :     ", y_test[idx])
exp = explainer_LR.explain_instance(X_test[idx], dec_tree.predict_proba,top_labels=5)
exp.show_in_notebook()

from tensorflow.keras.utils import to_categorical
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences

from tensorflow.keras.optimizers import Adam
from tensorflow.keras.models import Sequential
from tensorflow.keras.callbacks import EarlyStopping
from tensorflow.keras.layers import Dense, LSTM, Embedding, Bidirectional

temp = []
#Splitting pd.Series to list
data_to_list = df['text'].values.tolist()
for i in range(len(data_to_list)):
    temp.append(preprocess(data_to_list[i]))
data_words = list(sent_2_words(temp))
data = []
for i in range(len(data_words)):
    data.append(de_tokenize(data_words[i]))
print(data[:5])

from keras.preprocessing.text import Tokenizer
from keras.utils import pad_sequences
from keras import regularizers

max_words = 5000
max_len = max([len(t) for t in data])

tokenizer = Tokenizer(num_words=max_words)
tokenizer.fit_on_texts(data)
sequences = tokenizer.texts_to_sequences(data)
msgs = pad_sequences(sequences, maxlen=max_len)
print(msgs)

X_train = np.array(msgs[:15000])
X_test = np.array(msgs[15000:])
y_train = np.array(df['emotion_label'].values.tolist()[:15000])
y_test = np.array(df['emotion_label'].values.tolist()[15000:])

df.head()

vocabSize = len(tokenizer.index_word) + 1

!wget http://nlp.stanford.edu/data/glove.6B.zip

!unzip glove*.zip

# Read GloVE embeddings
path_to_glove_file = '/kaggle/working/glove.6B.200d.txt'
num_tokens = vocabSize
embedding_dim = 200 #latent factors or features  
hits = 0
misses = 0
embeddings_index = {}

# Load GloVe word embeddings from the specified file
with open(path_to_glove_file) as f:
    for line in f:
        word, coefs = line.split(maxsplit=1)
        coefs = np.fromstring(coefs, "f", sep=" ")
        embeddings_index[word] = coefs
print("Found %s word vectors." % len(embeddings_index))

# Initialize an embedding matrix for our neural network
embedding_matrix = np.zeros((num_tokens, embedding_dim))

# Assign pre-trained word vectors to our vocabulary
for word, i in tokenizer.word_index.items():
    embedding_vector = embeddings_index.get(word)
    if embedding_vector is not None:
        # Words found in the embedding index are assigned their respective vectors
        embedding_matrix[i] = embedding_vector
        hits += 1
    else:
        # Words not found in the embedding index are set to all-zeros
        # This includes the representation for "padding" and "OOV" (Out of Vocabulary)
        misses += 1
print("Converted %d words (%d misses)" % (hits, misses))

X_train.shape

# Define the sequential model
model = Sequential()

# Add an embedding layer with pre-trained weights and fixed trainable status
model.add(Embedding(vocabSize, 200, input_length=X_train.shape[1], weights=[embedding_matrix], trainable=False))

# Stack three Bidirectional LSTM layers with varying dropout rates
model.add(Bidirectional(LSTM(256, dropout=0.2, recurrent_dropout=0.2, return_sequences=True)))
model.add(Bidirectional(LSTM(128, dropout=0.3, recurrent_dropout=0.3, return_sequences=True)))
model.add(Bidirectional(LSTM(128, dropout=0.5, recurrent_dropout=0.5)))

# Add a dense layer with softmax activation for multi-class classification
model.add(Dense(6, activation='softmax'))

# Compile the model with sparse categorical cross-entropy loss and RMSprop optimizer
model.compile(loss='sparse_categorical_crossentropy', optimizer='rmsprop', metrics=['accuracy'])

# Display a summary of the model architecture
model.summary()

from keras.utils.vis_utils import plot_model
plot_model(model, show_shapes=True)

#to stop the training when the loss starts to increase
callback = EarlyStopping(
    monitor="val_loss",
    patience=4,
    restore_best_weights=True,
)

# Fit model
history = model.fit(X_train,
                    y_train,
                    validation_data=(X_test, y_test),
                    verbose=1,
                    batch_size=256,
                    epochs=15,
                    callbacks=[callback]
                   )